package org.fle4y.framework.net.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;
import org.fle4y.common.utils.configure.Configure;
import org.fle4y.common.utils.log.LogUtil;
import org.fle4y.framework.net.LogicHandler;
import org.fle4y.framework.net.Message;
import org.fle4y.framework.net.Protocal;
import org.fle4y.framework.net.Session;

/**
 * author : xiehui<br>
 * company : northking <br>
 * Create at: 2012-6-10<br>
 * @version : 1.0<br>
 * mailto:flexie@foxmail.com<br>
 *  Description :  会话实现
 */
public class AIOSession  implements Session
{
	private static final Logger log = LogUtil.getLogger(AIOSession.class.getName());
	private Long id;
	private ConcurrentHashMap<String,Object> attrs=new ConcurrentHashMap<String,Object>();
	private AsynchronousSocketChannel channel;
	/**协议*/
	private Protocal protocal;
	
	/**输入数据*/
	private final LinkedBlockingQueue<Message> inputQueue = new LinkedBlockingQueue<Message>();
	
	/**输入数据*/
	private final LinkedBlockingQueue<Message> outputQueue = new LinkedBlockingQueue<Message>();
	
	private ByteBuffer buffer;
	
	/**业务处理方式*/
	private LogicHandler logic;
	
	private Configure config;
	
	
	public AIOSession(Long id,AsynchronousSocketChannel channel,Protocal protocal,LogicHandler logic){
		this.id=id;
		this.channel=channel;
		this.protocal=protocal;
		this.logic=logic;
	}
	
	public void start(){
		try {
			if (channel.isOpen()) {
				SocketAddress remote=channel.getRemoteAddress();
				if(remote instanceof InetSocketAddress)
				{
					InetSocketAddress rInet=(InetSocketAddress)remote;
					log.info("来自客户端 IP : "+rInet.getHostName());
				}
				read();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	public Long getId() {
		return id;
	}
	public void init(Configure config) {
		try {
			this.config=config;
			channel.setOption(StandardSocketOptions.SO_KEEPALIVE,Boolean.parseBoolean(config.get("keep_alive").trim()));
			channel.setOption(StandardSocketOptions.SO_REUSEADDR,Boolean.parseBoolean(config.get("reuse_addr").trim()));
			channel.setOption(StandardSocketOptions.TCP_NODELAY,Boolean.parseBoolean(config.get("tcp_nodelay").trim()));
			
			/**
			 * 初始化的时候分配好一块内存,每次读取的重复使用这块内存.
			 * 针对传输文件这类特殊的需要保存每次过来的消息的情况在自定义的协议里面去处理
			 */
			int buffer_size=  Integer.parseInt(config.get("buffer_size").trim());
			buffer=ByteBuffer.allocate(buffer_size);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	public void open() {
		log.info("Open Session");
	}
	public void read(){
		buffer.clear();
		channel.read(buffer, this,new CompletionHandler<Integer,AIOSession>(){
			public void completed(Integer result, AIOSession session) {
				if(result>0){
					buffer.flip();
	    			Message msg=protocal.transfer(buffer);
	    			try {
						inputQueue.put(msg);
					} catch (InterruptedException e) {
	
						e.printStackTrace();
					}
	    			logic.doLogic(session, inputQueue,outputQueue);
	    			buffer.clear();
				}
				if(result<0){
					session.close();
					log.info("读取数据出错,关闭通道.");
					logic.endCallBack();
					return ;
				}
				read();
			}
			public void failed(Throwable exc, AIOSession session) {
			}
		});
	}
	public void add(Message msg) {
		try {
			outputQueue.put(msg);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	public void write() {
		try {
			Message msg;
			if(channel.isOpen()){
				while((msg = outputQueue.poll())!=null)
				{
					log.info("Session Id:"+id+"  写数据啦"+new String(msg.getContent().array(),Charset.forName("utf-8")));
					log.info(channel.toString());
					channel.write(msg.getContent(),this,new CompletionHandler<Integer,AIOSession>(){
						public void completed(Integer result, AIOSession attachment) {	
							if(result>0){
								log.info("write ok");
							}
						}
						public void failed(Throwable exc, AIOSession attachment) {
						}});
					msg.getContent().flip();
				}
			
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	public Object getAttribute(String key) {
		return attrs.get(key);
	}
	public void setAttribute(String key, Object value) {
		attrs.put(key, value);
	}
	
	public void close(){
	    try {
			channel.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	public Configure getConfig() {
		return config;
	}

	public LinkedBlockingQueue<Message> getOutputQueue() {
		return outputQueue;
	}



}
